package akka.mapreduce.actors

import scala.collection.mutable.ArrayBuffer

import akka.actor._
import akka.mapreduce._

class MapActor extends Actor {
  val STOP_WORDS = List("a", "am", "an", "and", "are", "as", "at", "be","do", "go", "if", "in", "is", "it", "of", "on", "the", "to", "this")

  def receive = {
    case message: String => sender ! evaluateExpression(message)
  }

  def evaluateExpression(line: String): MapData = MapData {
    line.split("""\s+""").foldLeft(ArrayBuffer.empty[WordCount]) {
      (index, word) => 
      if(! STOP_WORDS.contains(word.toLowerCase))
        index += WordCount(word.toLowerCase, 1)
      else
        index
    }
  }
}




